
#include <iostream>
#include <sstream>
#include <iomanip>

#include <boost/filesystem.hpp>
#include <boost/algorithm/string.hpp>

#include <boost/date_time/posix_time/posix_time.hpp>
using namespace boost::posix_time;


#include "bocat/common/ini.hpp"
#include "bocat/common/base64.hpp"
#include "bocat/common/time.hpp"
#include "bocat/common/logging.hpp"

#include "record.hpp"
#include "dubin.hpp"

namespace   sakcab   {


//static const char*	SNAPSHOT_ENCODE_base64 = "base64";


///////////////////////////////////////////////////////////////////////
//


Dubin::Dubin( const char* ini_filename, const char* ini_section )
		:   _config(ini_filename, ini_section),
            _filecell(_config.data_home, _config.data_file_rotate_size, _config.data_file_slim),
			_index_default_p( NULL ),
			_id( ini_section ),
			_meter( _id )
{
	initialize( ini_filename, ini_section );

	_when_last_snapshot = 0;
}

Dubin::~Dubin()
{
	finalize();
}


void
Dubin::initialize( const char* ini_filename, const char* ini_section )
{
	bocat::common::INIFile	ini( ini_filename );

	// get count of idx
	std::string	idx_count = ini[ ini_section ][ "index.count" ];
	if (  idx_count.size() < 1 ) {
		std::ostringstream	msg;
		msg << "No option 'index.count' for [" << ini_section << "] in " << ini_filename;
		throw std::runtime_error( msg.str() );
	}

	std::string	default_dbhome = ini[ ini_section ][ "index.home" ];
	if ( default_dbhome.size() < 1 ) {
		std::ostringstream	msg;
		msg << "No option 'index.home' for [" << ini_section << "] in " << ini_filename;
		throw std::runtime_error( msg.str() );
	}

	boost::filesystem::path	default_dbhome_path( default_dbhome );
	if ( !boost::filesystem::exists( default_dbhome_path ) ) {
		try {
			boost::filesystem::create_directories( default_dbhome_path );
		}
		catch ( std::exception& e ) {
			std::ostringstream	msg;
			msg << "fail to create index.home " << default_dbhome << " -- " << e.what();
			throw	std::runtime_error( msg.str() );
		}
	}

	// get default setting for idx
	std::string	idx_default_section = ini[ ini_section ][ "index.setting" ];
	// override default dbhome
	ini[ idx_default_section ][ "index.home" ] = default_dbhome;

	std::ostringstream	stream_ini;
	stream_ini << ini[ idx_default_section ];
	_idx_default_settings = stream_ini.str();

	std::string	default_section = ini[ idx_default_section ][ "index.section" ];
	if ( default_section.size() < 1 ) {
		std::ostringstream	msg;
		msg << "No option 'index.section' for [" << idx_default_section << "] in " << ini_filename;
		throw std::runtime_error( msg.str() );
	}

	// create default bin
	std::string	default_db_filename = ini[ ini_section ][ "index.dbfile" ];
	if ( default_db_filename.size() < 1 ) {
		std::ostringstream	msg;
		msg << "No option 'index.dbfile' for [" << ini_section << "] in " << ini_filename;
		throw std::runtime_error( msg.str() );
	}
	std::string	default_ini_filename = default_db_filename + ".ini";
	boost::filesystem::path	default_ini_filepath( default_ini_filename );
	if ( !boost::filesystem::exists( default_ini_filepath ) ) {
		std::ostringstream	stream;
		stream << ini[ ini_section ];

		std::ofstream	ofs( default_ini_filename.c_str() );
		if ( !ofs ) {
			std::ostringstream	msg;
			msg << "fail to create " << default_ini_filename;
			throw std::runtime_error( msg.str() );
		}

		ofs << "[" << default_section << "]" << std::endl
			<< stream.str() << std::endl
			<< std::endl;

		ofs.close();
	}
	_index_default_p = new BinIndex( default_ini_filename.c_str(), default_section.c_str() );

	// create other bin
	boost::trim( idx_count );
	unsigned	idx_count_i = boost::lexical_cast< unsigned >( idx_count.c_str() );
	for (unsigned i=0; i<idx_count_i; ++i) {
		std::ostringstream	stream_db;
		stream_db << default_dbhome << "/" << std::setw(idx_count.size()) << std::setfill('0') << i << ".db";

		std::string	idx_db = stream_db.str();
		std::string	idx_ini = idx_db + ".ini";

		boost::filesystem::path	ini_path( idx_ini );
		if ( !boost::filesystem::exists( ini_path ) ) {
			std::ofstream	ofs( idx_ini.c_str() );
			if ( !ofs ) {
				std::ostringstream	msg;
				msg << "fail to create " << idx_ini;
				throw std::runtime_error( msg.str() );
			}

			ofs << "[" << default_section << "]" << std::endl
				<< "index.dbfile = " << idx_db << std::endl
				<< _idx_default_settings << std::endl
				<< std::endl;

			ofs.close();
		}

		index_ptr	idx_p = new BinIndex( idx_ini.c_str(), default_section.c_str() );
		_index_bins.push_back( idx_p );
	}

	// create snapshot home, if necessary
	if ( _config.index_snap_home.size() > 0 ) {
		boost::filesystem::path	snap_home_path( _config.index_snap_home );
		if ( !boost::filesystem::exists( snap_home_path ) ) {
			try {
				boost::filesystem::create_directories( snap_home_path );
			}
			catch ( std::exception& e ) {
				std::ostringstream	msg;
				msg << "fail to create index.snap.home " << _config.index_snap_home << " -- " << e.what();
				throw	std::runtime_error( msg.str() );
			}
		}
	}
}


void
Dubin::finalize( void )
{
	bocat::common::Logger*  logger_p = bocat::common::getLogger();

	// snpshot
	logger_p->info(__LINE__, __FILE__, "Dubin : make indexes checkpoint ...");
	this->snapshot();

	// release all index
	logger_p->info(__LINE__, __FILE__, "Dubin : close indexes ...");
	for (unsigned i=0; i < _index_bins.size(); ++i ) {
		index_ptr	p = _index_bins[ i ];
		delete p;
	}
	_index_bins.clear();

	if (NULL != _index_default_p) {
		delete _index_default_p;

		_index_default_p = NULL;
	}
	logger_p->info(__LINE__, __FILE__, "Dubin : done.");
}


///////////////////////////////////////////////////////////////////////
//

static	void
composeTimestampLocation( const Timestamp& ts, const RecordLocation& loc, std::string& value )
{
	value.reserve( sizeof(Timestamp) + sizeof(RecordLocation) );
	value.append( (const char*)&ts,  sizeof(Timestamp) );
	value.append( (const char*)&loc, sizeof(RecordLocation) );
}

static	void
decomposeTimestampLocation( const std::string& value, Timestamp& ts, RecordLocation& loc )
{
	if( (sizeof(Timestamp) + sizeof(RecordLocation)) != value.size() ) {
		std::ostringstream	msg;
		msg << "bad format for timestamp and location data";
		throw	std::runtime_error( msg.str() );
	}

	const char*	p = value.c_str();
	memcpy( (void *)&ts, p, sizeof(Timestamp) );
	memcpy( (void *)&loc, p + sizeof(Timestamp), sizeof(RecordLocation) );
}

///////////////////////////////////////////////////////////////////////
//

int 
Dubin::insert( const std::string& kpin,
         const std::string& key,
         const std::string& data )
{
	boost::mutex::scoped_lock   _scoped_lock( _mutex );

	ptime   time_start = microsec_clock::local_time();

	index_ptr	index_p = getIndex( kpin, key );

    // STEP. check if key exists
    std::string skey = getSkeyByKpin( kpin );
    RecordLocation  prev_loc;
    RecordLocation  loc;
    try {
		std::string		tsnloc;
        index_p->get( kpin, skey, key, tsnloc );

		if ( tsnloc.size() > 0 ) {
			Timestamp	ts( (time_t)0 );
			decomposeTimestampLocation( tsnloc, ts, loc );
		}
    }   
    catch( std::exception& e ) { 
        std::ostringstream  msg;
        msg << "fail to check existence for key '" << key << "' -- " << e.what();
        throw std::runtime_error( msg.str() );
    }   

    // STEP. if key exists, delete the old one
    std::string skey_b64 = bocat::common::base64_encode( skey );
    std::string kpin_b64 = bocat::common::base64_encode( kpin );

    if ( !loc.is_null() ) { 
		Timestamp	ts;
	
        _filecell.erase( skey_b64, kpin_b64, key, ts, loc );
    }   

    // STEP. insert the record
    //       . put data into file cell
    //       . put location into index

	std::string	key_stored;
	if ( _config.data_key_encode_base64_b ) {
		key_stored = bocat::common::base64_encode( key );
	}
	else {
		key_stored = key;
	}
	std::string	data_stored;
	if ( _config.data_value_encode_base64_b ) {
		data_stored = bocat::common::base64_encode( data );
	}
	else {
		data_stored = data;
	}

	Timestamp	ts;
	
    loc.clear();
    _filecell.put( skey_b64, kpin_b64, key_stored, data_stored, ts, loc, prev_loc );

	std::string	value;
	composeTimestampLocation( ts, loc, value );

    index_p->put( kpin, skey, key, value );

	ptime   time_stop = microsec_clock::local_time();
	long	time_last = getTimeDuration( time_start, time_stop );
	_meter.countInsert( time_last );

    return  0;
}


int 
Dubin::insertAppend( const std::string& kpin,
         const std::string& key,
         const std::string& data )
{
	boost::mutex::scoped_lock   _scoped_lock( _mutex );

	ptime   time_start = microsec_clock::local_time();

	index_ptr	index_p = getIndex( kpin, key );

    // STEP. check if key exists
    std::string skey = getSkeyByKpin( kpin );
    RecordLocation  prev_loc;
    try {
		std::string		tsnloc;
        index_p->get( kpin, skey, key, tsnloc );

		if ( tsnloc.size() > 0 ) {
			Timestamp	ts( (time_t)0 );
			decomposeTimestampLocation( tsnloc, ts, prev_loc );
		}
    }
    catch( std::exception& e ) {
        std::ostringstream  msg;
        msg << "fail to check existence for key '" << key << "' -- " << e.what();
        throw std::runtime_error( msg.str() );
    }

    // STEP. insert the record
    //       . put data into file cell
    //       . put location into index
    std::string skey_b64 = bocat::common::base64_encode( skey );
    std::string kpin_b64 = bocat::common::base64_encode( kpin );


	std::string	key_stored;
	if ( _config.data_key_encode_base64_b ) {
		key_stored = bocat::common::base64_encode( key );
	}
	else {
		key_stored = key;
	}
	std::string	data_stored;
	if ( _config.data_value_encode_base64_b ) {
		data_stored = bocat::common::base64_encode( data );
	}
	else {
		data_stored = data;
	}

	Timestamp	ts;
    RecordLocation  loc;

    _filecell.put( skey_b64, kpin_b64, key_stored, data_stored, ts, loc, prev_loc );

	std::string	value;
	composeTimestampLocation( ts, loc, value );

    index_p->put( kpin, skey, key, value );

	ptime   time_stop = microsec_clock::local_time();
	long	time_last = getTimeDuration( time_start, time_stop );
	_meter.countInsert( time_last );

    return  0;
}


int 
Dubin::insertIgnore( const std::string& kpin,
         const std::string& key,
         const std::string& data )
{
	boost::mutex::scoped_lock   _scoped_lock( _mutex );

	ptime   time_start = microsec_clock::local_time();

	index_ptr	index_p = getIndex( kpin, key );

    // STEP. check if key exists
    std::string skey = getSkeyByKpin( kpin );
    RecordLocation  loc;
    try {
		std::string		tsnloc;
        index_p->get( kpin, skey, key, tsnloc );

		if ( tsnloc.size() > 0 ) {
			Timestamp	ts( (time_t)0 );
			decomposeTimestampLocation( tsnloc, ts, loc );
		}
    }
    catch( std::exception& e ) {
        std::ostringstream  msg;
        msg << "fail to check existence for key '" << key << "' -- " << e.what();
        throw std::runtime_error( msg.str() );
    }

    // STEP. if key exists, ignore it
    if ( !loc.is_null() ) {
        return 0;
    }

    // STEP. insert the record
    //       . put data into file cell
    //       . put location into index
    std::string skey_b64 = bocat::common::base64_encode( skey );
    std::string kpin_b64 = bocat::common::base64_encode( kpin );

	std::string	key_stored;
	if ( _config.data_key_encode_base64_b ) {
		key_stored = bocat::common::base64_encode( key );
	}
	else {
		key_stored = key;
	}
	std::string	data_stored;
	if ( _config.data_value_encode_base64_b ) {
		data_stored = bocat::common::base64_encode( data );
	}
	else {
		data_stored = data;
	}

	Timestamp	ts;
    RecordLocation  prev_loc;
    loc.clear();

    _filecell.put( skey_b64, kpin_b64, key_stored, data_stored, ts, loc, prev_loc );

	std::string	value;
	composeTimestampLocation( ts, loc, value );

    index_p->put( kpin, skey, key, value );

	ptime   time_stop = microsec_clock::local_time();
	long	time_last = getTimeDuration( time_start, time_stop );
	_meter.countInsert( time_last );


    return  0;
}


int 
Dubin::find( const std::string& kpin,
         const std::string& key,
               std::string& data )
{   
	boost::mutex::scoped_lock   _scoped_lock( _mutex_read );

	ptime   time_start = microsec_clock::local_time();

	index_ptr	index_p = getIndex( kpin, key );

    // STEP. check if key exists
    std::string skey = getSkeyByKpin( kpin );
    RecordLocation  loc;
    try {
		std::string		tsnloc;
        index_p->get( kpin, skey, key, tsnloc );

		if ( tsnloc.size() > 0 ) {
			Timestamp	ts( (time_t)0 );
			decomposeTimestampLocation( tsnloc, ts, loc );
		}
    }
    catch( std::exception& e ) {
        std::ostringstream  msg;
        msg << "fail to check existence for key '" << key << "' -- " << e.what();
        throw std::runtime_error( msg.str() );
    }

    // STEP. if key does NOT exist, just return
    if ( loc.is_null() ) {
        return 0;
    }

    // STEP. get the record
    //       . get data from file cell
    std::string skey_b64 = bocat::common::base64_encode( skey );
    std::string kpin_b64 = bocat::common::base64_encode( kpin );

    RecordLocation  prev_loc;

	std::string	data_stored;
	std::string	key_stored;
	if ( _config.data_key_encode_base64_b ) {
		key_stored = bocat::common::base64_encode( key );
	}
	else {
		key_stored = key;
	}

	Timestamp	ts( (time_t)0 );
    _filecell.get( skey_b64, kpin_b64, key_stored, data_stored, ts, loc, prev_loc );

	if ( _config.data_value_encode_base64_b ) {
		data = bocat::common::base64_decode< std::string >( data_stored );
	}
	else {
		data = data_stored;
	}

	ptime   time_stop = microsec_clock::local_time();
	long	time_last = getTimeDuration( time_start, time_stop );
	_meter.countFind( time_last );


    return  data.size();
}


int
Dubin::find( const std::string& kpin,
             const std::string& key,
                   std::vector< std::string >& all_data,
                   u_int32_t limit )
{
    boost::mutex::scoped_lock   _scoped_lock( _mutex_read );

	ptime   time_start = microsec_clock::local_time();

	index_ptr	index_p = getIndex( kpin, key );

    // STEP. check if key exists
    std::string skey = getSkeyByKpin( kpin );
    RecordLocation  loc;
    try {
		std::string		tsnloc;
        index_p->get( kpin, skey, key, tsnloc );

		if ( tsnloc.size() > 0 ) {
			Timestamp	ts( (time_t)0 );
			decomposeTimestampLocation( tsnloc, ts, loc );
		}
    }
    catch( std::exception& e ) {
        std::ostringstream  msg;
        msg << "fail to check existence for key '" << key << "' -- " << e.what();
        throw std::runtime_error( msg.str() );
    }

    // STEP. if key does NOT exist, just return
    if ( loc.is_null() ) {
		//std::string skey_b64 = bocat::common::base64_encode( skey );
		//std::string kpin_b64 = bocat::common::base64_encode( kpin );
		//std::cerr << __FILE__ << __LINE__ << " NOT FOUND key:" << key << " kpin-b64:" << kpin_b64 << std::endl;
        return 0;
    }

    // STEP. get the record
    //       . get data from file cell
    std::string skey_b64 = bocat::common::base64_encode( skey );
    std::string kpin_b64 = bocat::common::base64_encode( kpin );

	std::string	data_stored;
	std::string	key_stored;
	if ( _config.data_key_encode_base64_b ) {
		key_stored = bocat::common::base64_encode( key );
	}
	else {
		key_stored = key;
	}

    while ( true ) {
        std::string data;
        RecordLocation  prev_loc;
		Timestamp	ts( (time_t)0 );
        _filecell.get( skey_b64, kpin_b64, key_stored, data_stored, ts, loc, prev_loc );

		if ( _config.data_value_encode_base64_b ) {
			data = bocat::common::base64_decode< std::string >( data_stored );
		}
		else {
			data = data_stored;
		}
        all_data.push_back( data );

        if ( prev_loc.is_null() ) {
            break;
        }
        loc = prev_loc;

        if ( (limit > 0) && (limit <= all_data.size()) ) {
            break;
        }
    }

	ptime   time_stop = microsec_clock::local_time();
	long	time_last = getTimeDuration( time_start, time_stop );
	_meter.countFind( time_last );

    return  all_data.size();
}



///////////////////////////////////////////////////////////////////////
//

int 
Dubin::erase( const std::string& kpin,
           const std::string& key )
{
	boost::mutex::scoped_lock   _scoped_lock( _mutex );

	ptime   time_start = microsec_clock::local_time();

	index_ptr	index_p = getIndex( kpin, key );

    // STEP. check if key exists
    std::string skey = getSkeyByKpin( kpin );
    RecordLocation  loc;
    try {
		std::string		tsnloc;
        index_p->get( kpin, skey, key, tsnloc );

		if ( tsnloc.size() > 0 ) {
			Timestamp	ts( (time_t)0 );
			decomposeTimestampLocation( tsnloc, ts, loc );
		}
    }
    catch( std::exception& e ) {
        std::ostringstream  msg;
        msg << "fail to check existence for key '" << key << "' -- " << e.what();
        throw std::runtime_error( msg.str() );
    }

    // STEP. if key does NOT exist, just return
    if ( loc.is_null() ) {
        return 0;
    }

    // STEP. erase the record
    //       . erase data from file cell
    //       . erase index from file cell
    std::string skey_b64 = bocat::common::base64_encode( skey );
    std::string kpin_b64 = bocat::common::base64_encode( kpin );

	std::string	key_stored;
	if ( _config.data_key_encode_base64_b ) {
		key_stored = bocat::common::base64_encode( key );
	}
	else {
		key_stored = key;
	}

	Timestamp	ts;
    _filecell.erase( skey_b64, kpin_b64, key_stored, ts, loc );
    index_p->erase( kpin, skey, key );

	ptime   time_stop = microsec_clock::local_time();
	long	time_last = getTimeDuration( time_start, time_stop );
	_meter.countErase( time_last );

    return  0;
}


///////////////////////////////////////////////////////////////////////
//

int
Dubin::insertAppend( const std::vector< std::string >& kpin_v,
					 const std::vector< std::string >& key_v,    
					 const std::vector< std::string >& data_v )
{
	boost::mutex::scoped_lock   _scoped_lock( _mutex );


	for (size_t k=0; k<key_v.size(); ++k) {
		const std::string&	kpin = kpin_v[ k ];
		const std::string&	key  = key_v[ k ];
		const std::string&	data = data_v[ k ];

		index_ptr	index_p = getIndex( kpin, key );

		// STEP. check if key exists
		std::string skey = getSkeyByKpin( kpin );
		RecordLocation  prev_loc;
		try {
			std::string		tsnloc;
			index_p->get( kpin, skey, key, tsnloc );

			if ( tsnloc.size() > 0 ) {
				Timestamp	ts( (time_t)0 );
				decomposeTimestampLocation( tsnloc, ts, prev_loc );
			}
		}
		catch( std::exception& e ) {
			std::ostringstream  msg;
			msg << "fail to check existence for key '" << key << "' -- " << e.what();
			throw std::runtime_error( msg.str() );
		}

		// STEP. insert the record
		//       . put data into file cell
		//       . put location into index
		std::string skey_b64 = bocat::common::base64_encode( skey );
		std::string kpin_b64 = bocat::common::base64_encode( kpin );

		RecordLocation  loc;

		std::string	key_stored;
		if ( _config.data_key_encode_base64_b ) {
			key_stored = bocat::common::base64_encode( key );
		}
		else {
			key_stored = key;
		}
		std::string	data_stored;
		if ( _config.data_value_encode_base64_b ) {
			data_stored = bocat::common::base64_encode( data );
		}
		else {
			data_stored = data;
		}

		Timestamp	ts;
		_filecell.put( skey_b64, kpin_b64, key_stored, data_stored, ts, loc, prev_loc );

		std::string	value;
		composeTimestampLocation( ts, loc, value );

		index_p->put( kpin, skey, key, value );
	}

    return  0;
}


///////////////////////////////////////////////////////////////////////
//


int
Dubin::slim( void )
{
    boost::mutex::scoped_lock   _scoped_lock( _mutex );

    IndexIterator*  it_p = NULL;
    int     count_slimmed = 0;

    bocat::common::Logger*  logger_p = bocat::common::getLogger();

    while( (it_p = _filecell.getSlimIndexIterator()) != NULL ) { 
        try {
            logger_p->info(__LINE__, __FILE__, " slimming %s ...", _filecell.getSlimmingFile().c_str());

			std::string kpin, skey, key;
			int		fileno_slimming = _filecell.getSlimmingFileNo();
			int		count_slim_each_file = 0;
			Timestamp	ts( (time_t)0 );
			std::string	op;

			while ( it_p->next(kpin, skey, key, ts, op) > 0 ) {	
				index_ptr	index_p = getIndex( kpin, key );

				// STEP. check if key exists
				std::string skey = getSkeyByKpin( kpin );
				RecordLocation  loc;
				Timestamp	index_ts( (time_t)0 );
				try {
					std::string		tsnloc;
					index_p->get( kpin, skey, key, tsnloc );

					if ( tsnloc.size() > 0 ) {
						decomposeTimestampLocation( tsnloc, index_ts, loc );
					}
				}
				catch( std::exception& e ) {
					// just ignore ...
					//std::ostringstream  msg;
					//msg << "fail to check existence for key '" << key << "' -- " << e.what();
					////throw std::runtime_error( msg.str() );
					//logger_p->error( __LINE__,__FILE__, "SLIM ERROR : %s", msg.str().c_str() );
					continue;
				}

				// check if fileno matched
				if ( fileno_slimming != loc.fileno ) {
					continue;
				}

				if ( FileCell::OP_PUT == op ) {
					if ( index_ts != ts ) {
						continue;
					}
				}
				else if ( FileCell::OP_DEL == op ) {
					if ( ts < index_ts ) {
						// DEL op occurs before index-update time
						continue;
					}
				}

				//std::string skey_b64 = bocat::common::base64_encode( skey );
				//std::string kpin_b64 = bocat::common::base64_encode( kpin );
				//_filecell.erase( skey_b64, kpin_b64, key, loc );
				index_p->erase( kpin, skey, key );

				++count_slim_each_file;
                ++count_slimmed; 
			}
            logger_p->info(__LINE__,__FILE__," slim %s done (%d records slimmed).", _filecell.getSlimmingFile().c_str(), count_slim_each_file);
        }
        catch( std::exception& e ) {
            logger_p->info(__LINE__,__FILE__," slim %s failed.", _filecell.getSlimmingFile().c_str());
            logger_p->error(__LINE__,__FILE__," slim exception : %s", e.what());
            logger_p->error(__LINE__,__FILE__," slim %s failed.", _filecell.getSlimmingFile().c_str());

            _filecell.freeSlimIndexIterator( it_p, false );
            continue;
        }       

        _filecell.freeSlimIndexIterator( it_p );
    }       

	// optimize index
	logger_p->info(__LINE__, __FILE__, " optimizing index %s ...", _id.c_str());
	for (unsigned i=0; i < _index_bins.size(); ++i ) {
		logger_p->info(__LINE__, __FILE__, "  - optimizing #%u index ...", i);
		index_ptr	p = _index_bins[ i ];

		p->optimize();
	}

	if (NULL != _index_default_p) {
		logger_p->info(__LINE__, __FILE__, "  - optimizing default index ...");
		_index_default_p->optimize();
	}
	logger_p->info(__LINE__, __FILE__, " optimize index %s done.", _id.c_str());

	// snap-shot
	this->snapshot();

    return  count_slimmed;
}


int
Dubin::snapshot( void )
{
    bocat::common::Logger*  logger_p = bocat::common::getLogger();

	time_t	now = ::time( NULL );
	if ( (_config.index_snap_home.size() > 0) && ((now - _when_last_snapshot) > _config.index_snap_interval) ) {
		logger_p->info(__LINE__, __FILE__, " snapshoting index %s ...", _id.c_str());
		for (unsigned i=0; i < _index_bins.size(); ++i ) {
			char	filename[ 512 ], fn_prefix[ 32 ];
			if ( _index_bins.size() < 10 ) {
				snprintf( filename, sizeof(filename) - 1, "%s/%u.snap-%lu", _config.index_snap_home.c_str(), i, now );
				snprintf( fn_prefix, sizeof(fn_prefix) - 1, "%u.snap-", i );
			}
			else {
				snprintf( filename, sizeof(filename) - 1, "%s/%02u.snap-%lu", _config.index_snap_home.c_str(), i, now );
				snprintf( fn_prefix, sizeof(fn_prefix) - 1, "%02u.snap-", i );
			}
			logger_p->info(__LINE__, __FILE__, "  - snapshoting --> %s ...", filename);
			index_ptr	p = _index_bins[ i ];

			int	count_records = p->snapshot( filename, _config.index_snap_key_encode.c_str(), _config.index_snap_value_encode.c_str() );

			checkSnapshotCopies( _config.index_snap_home.c_str(), fn_prefix, _config.index_snap_copies );

			logger_p->info(__LINE__, __FILE__, "    snapshot %s done (%d records).", filename, count_records);
		}

		if (NULL != _index_default_p) {
			char	filename[ 512 ], fn_prefix[ 32 ];
			snprintf( filename, sizeof(filename) - 1, "%s/default.snap-%lu", _config.index_snap_home.c_str(), now );
			snprintf( fn_prefix, sizeof(fn_prefix) - 1, "default.snap-" );
			logger_p->info(__LINE__, __FILE__, "  - snapshoting --> %s ...", filename);

			int	count_records = _index_default_p->snapshot( filename, _config.index_snap_key_encode.c_str(), _config.index_snap_value_encode.c_str() );

			checkSnapshotCopies( _config.index_snap_home.c_str(), fn_prefix, _config.index_snap_copies );

			logger_p->info(__LINE__, __FILE__, "    snapshot %s done (%d records).", filename, count_records);
		}
		logger_p->info(__LINE__, __FILE__, " snapshot index %s done.", _id.c_str());

		_when_last_snapshot = ::time(NULL);
	}

	return	0;
}


///////////////////////////////////////////////////////////////////////
//


const std::string   
Dubin::getSkeyByKpin( const std::string& kpin )
{
	char	skey[32];
	u_int32_t	skey_size = _config.index_skey_size;

	memset(skey, 0, sizeof(skey));
	if ( kpin.size() < skey_size ) {
		skey_size = kpin.size();
	}

	memcpy( skey, kpin.c_str() + (kpin.size() - skey_size), skey_size );

	return	std::string( skey, skey_size );
}


Dubin::index_ptr
Dubin::getIndex( const std::string& kpin, const std::string key )
{
	if ( _index_bins.size() < 1 ) {
		return	_index_default_p;
	}

	// assert( kpin.size() == 16 );

	const char*	p = kpin.c_str();
	size_t	s = kpin.size();
	if ( s < 16 ) {
		std::ostringstream	msg;
		msg << "BAD kpin (size=" << s << ") for KEY: " << key;

		throw std::runtime_error( msg.str() );
	}

	int	x =  (p[7] << 8) + p[8];
	return	_index_bins[ x % _index_bins.size() ];
}

///////////////////////////////////////////////////////////////////////
//

}   //// namespace sakcab

